package tcpsubpub

import (
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/gotask"
	"gitee.com/ymofen/gobase/subpub"
	"gitee.com/ymofen/kplbase/core/knet"
	"io"
	"sync"
	"sync/atomic"
	"time"
)

type TcpClientNode struct {
	disflag          int32
	closedflag       int32
	sessionSN        int32
	connectcnt       int32
	startedflag      int32
	lk               sync.RWMutex
	lastRecvT        time.Time
	checkRecvTimeOut time.Duration
	addr             string
	proxydialAddr    string
	sendStrConnected string
	disReason        string
	maxonline        int
	tcpcli           *knet.TcpCliConn
	ex               *subpub.Subchannel
}

func (this *TcpClientNode) checkStart() error {
	this.lk.Lock()
	defer this.lk.Unlock()
	if this.startedflag == 1 {
		return nil
	}

	this.startedflag = 1

	tcpcli := knet.NewTcpCliConn()
	tcpcli.ConfigRemoteAddr(this.addr)
	if len(this.proxydialAddr) > 0 {
		tcpcli.ConfigProxyDialAddr(this.proxydialAddr)
	}

	tcpcli.OnConnectFail = func(sender *knet.TcpCliConn, err error) {
		this.ex.Pub("evt", 0, subpub.ACTION_CONNECTFAIL, err)
	}

	atomic.StoreInt32(&this.closedflag, 0)
	this.tcpcli = tcpcli
	go this.innerHandleConnFunc(tcpcli)

	gotask.DefaultPatrolTask().AddTask(this, time.Second*10, func(id interface{}, args ...interface{}) {
		go this.checkRecv()
	})

	return nil
}

func (this *TcpClientNode) checkClose() error {
	this.lk.Lock()
	defer this.lk.Unlock()
	if this.startedflag == 0 {
		return io.ErrClosedPipe
	}

	tcpcli := this.tcpcli
	if tcpcli == nil {
		return io.ErrClosedPipe
	}

	atomic.StoreInt32(&this.closedflag, 1)

	return tcpcli.Close()
}

func (this *TcpClientNode) Close() error {
	err := this.checkClose()
	this.ex.Pub("notify-closed", 0, "请求关闭")
	return err
}

func (this *TcpClientNode) reqDis(reason string) {
	if atomic.CompareAndSwapInt32(&this.disflag, 0, 1) {
		this.lk.RLock()
		tcpcli := this.tcpcli
		this.lk.RUnlock()

		if tcpcli != nil {
			tcpcli.Close()
			this.disReason = reason
			this.ex.Pub("evt", 0, subpub.ACTION_DISCONNECTED, reason)
		}
	}
}

func (this *TcpClientNode) checkRecv() {
	if this.checkRecvTimeOut > 0 {
		this.lk.RLock()
		started := this.startedflag
		tcpcli := this.tcpcli
		this.lk.RUnlock()

		if started == 0 || tcpcli == nil {
			return
		}

		if time.Since(this.lastRecvT) > this.checkRecvTimeOut {
			this.reqDis("checkTimeOut")
		}
	}
}

func (this *TcpClientNode) innerHandleConnFunc(tcpcli *knet.TcpCliConn) {
	if gobase.GoFunCatchException {
		defer gobase.DeferCatchPanic()
	}

	for atomic.LoadInt32(&this.closedflag) == 0 {
		r := tcpcli.CheckConnect()
		if r == -1 {
			atomic.StoreInt32(&this.disflag, 0)
			time.Sleep(time.Second)
			continue
		} else if r == 1 {
			s0 := this.sendStrConnected
			if len(s0) > 0 {
				tcpcli.Write([]byte(s0))
			}
			this.lastRecvT = time.Now()
			this.ex.Pub("evt", 0, subpub.ACTION_CONNECTED, tcpcli.GetRemoteAddr(), s0)
			atomic.StoreInt32(&this.disflag, 0)
		}

		buf := make([]byte, 4096)
		n, e1 := tcpcli.Read(buf)
		if e1 != nil {
			this.reqDis(e1.Error())
			time.Sleep(time.Second)
			continue
		}
		this.lastRecvT = time.Now()
		this.ex.Pub("buf", 0, buf[:n])
	}
}

var (
	DefaultTcpClientGroup = gobase.NewConnectObjectRefStorage(newTcpClientNode)
)

func newTcpClientNode(connstr string) interface{} {
	conf := gobase.NewStrMap()
	conf.URLFormDecode(connstr)
	node := &TcpClientNode{addr: conf.StringByName("addr", ""), proxydialAddr: conf.StringByName("proxydial", "")}
	node.sendStrConnected = conf.StringByName("sendstrconnected", "")
	node.checkRecvTimeOut = time.Duration(conf.IntByName("checkrecvtimeout", 0)) * time.Second
	node.ex = subpub.NewSubchannel()
	return node
}
